package com.deepbreath.spider.pm25in;

import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.lang3.StringUtils;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;

import com.deepbreath.entity.Position;
import com.deepbreath.spider.AbstractProcessor;

import com.deepbreath.spider.HanyuPinyinHelper;

import com.deepbreath.spider.pm25in.PositionProcessor;

import com.alibaba.fastjson.JSON;
import com.deepbreath.entity.CitySite;
import us.codecraft.webmagic.Page;
import us.codecraft.webmagic.Request;
import us.codecraft.webmagic.Site;
import us.codecraft.webmagic.Spider;
import us.codecraft.webmagic.processor.PageProcessor;

public class PositionProcessor extends AbstractProcessor<List<CitySite>> implements PageProcessor {

	public static final String NET_URL = "http://www.pm25.in/";
	public static final String CITY_CODE_KEY = "city_code";
	public static final String SEQ = "-";
	private Site site = null;
	private Spider spider = null;

	private Map<String, Position> DATA_MAP = new ConcurrentHashMap<String, Position>();

	public Collection<Position> getDATA_MAP() {
		return DATA_MAP.values();
	}

	@Override
	public void process(Page page) {
		try {
			Document doc = page.getHtml().getDocument();
			String city = doc.select("div.city_name").text();

			// 读取更新时间
			String time = doc.select("div.live_data_time>p").text();
			time = time.substring(time.indexOf("：") + 1);
			Timestamp ts = new Timestamp(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(time).getTime());

			// 读取当前城市的综合数据
			List<String> citydatas = doc.select("div.span12.data>div.span1>div.value").eachText();
			if (citydatas.size() >= 8) {
				Position pcity = new Position();
				pcity.setCity(city);
				pcity.setPositionName(city);// 监测站点名称
				pcity.setTime(ts);

				String affect = doc.select("div.affect").text().replace(" ", "");
				affect = affect.substring(affect.indexOf("：") + 1);
				String action = doc.select("div.action").text().replace(" ", "");
				action = action.substring(action.indexOf("：") + 1);
				String live_data_unit = doc.select("div.live_data_unit").text().replace(" ", "");
				live_data_unit = live_data_unit.substring(live_data_unit.indexOf("：") + 1);
				String primary_pollutant = doc.select("div.primary_pollutant").text().trim();
				primary_pollutant = primary_pollutant.substring(primary_pollutant.indexOf("：") + 1);
				pcity.setAffect(StringUtils.isBlank(affect) ? "-" : affect);// 对健康影响情况
				pcity.setAction(StringUtils.isBlank(action) ? "-" : action);// 建议采取的措施
				pcity.setLiveDataUnit(StringUtils.isBlank(live_data_unit) ? "-" : live_data_unit);// 数值单位
				pcity.setPrimaryPollutant(StringUtils.isBlank(primary_pollutant) ? "-" : primary_pollutant);// 首要污染物

				pcity.setQuality(doc.select("div.span12.avg").select("div.level>h4").text().trim());// 空气质量(指数类别)
				pcity.setAqi(Position.toDouble(citydatas.get(0), null));// AQI
				pcity.setPm25(Position.toDouble(citydatas.get(1), null));// PM2.5(细颗粒物)
				pcity.setPm10(Position.toDouble(citydatas.get(2), null));// PM10(可吸入颗粒物)
				pcity.setCo(Position.toDouble(citydatas.get(3), null));// CO(一氧化碳)
				pcity.setNo2(Position.toDouble(citydatas.get(4), null));// NO2(二氧化氮)
				pcity.setO3(Position.toDouble(citydatas.get(5), null));// O3(臭氧1小时平均)
				pcity.setO38h(Position.toDouble(citydatas.get(6), null));// O3(臭氧8小时平均)
				pcity.setSo2(Position.toDouble(citydatas.get(7), null));// SO2(二氧化硫)

				StringBuffer sb = new StringBuffer();
				sb.append(new SimpleDateFormat("yyyyMMddHHmmss").format(ts)).append(SEQ);
				sb.append(page.getRequest().getHeaders().get(CITY_CODE_KEY));
				pcity.setId(sb.toString());
				DATA_MAP.put(pcity.getId(), pcity);
			}

			// 读取监测数据
			Elements elements = doc.select("table#detail-data>tbody>tr");
			for (Element element : elements) {
				Elements tds = element.select("td");
				if (tds.size() != 11) {
					throw new RuntimeException("非法格式:" + element.html());
				}

				List<String> list = new ArrayList<String>();
				for (Element td : tds) {
					list.add(td.text());
				}
				Position position = Position.build(list);
				position.setTime(ts);
				position.setCity(city);

				StringBuffer sb = new StringBuffer();
				SimpleDateFormat RSDF = new SimpleDateFormat("yyyyMMddHHmmss");
				sb.append(RSDF.format(ts)).append(SEQ);
				sb.append(HanyuPinyinHelper.toHanyuPinyin(list.get(0)));
				position.setId(sb.toString());

				DATA_MAP.put(position.getId(), position);

			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public Site getSite() {
		return site;
	}

	@Override
	public void initialize(List<CitySite> citys) {
		site = Site.me().setSleepTime(3000).setTimeOut(3000).setRetryTimes(5);
		Request[] requests = new Request[citys.size()];
		for (int i = 0; i < citys.size(); i++) {
			CitySite citySite = citys.get(i);
			requests[i] = new Request(NET_URL + citySite.getId());
			requests[i].addHeader(CITY_CODE_KEY, citySite.getId());
		}
		spider = Spider.create(this).addRequest(requests).thread(5);
	}

	@Override
	public void start() {
		spider.run();
	}

	@Override
	public void shutdown() {
		spider.close();
	}

	public static void main(String[] args) {
		PositionProcessor processor = new PositionProcessor();
		List<CitySite> citys = new ArrayList<CitySite>();
		citys.add(new CitySite(null, null, null, null, "anshan", null));
		processor.initialize(citys);
		processor.start();
		System.out.println(JSON.toJSONString(processor.getDATA_MAP()));
	}
}
